package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.util.ExcelUtils;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.pojo.Point0F37;
import com.navinfo.tripanalysis.pojo.LoadDataParam;
import com.navinfo.tripanalysis.pojo.Point;
import com.navinfo.tripanalysis.service.Load0F37DataService;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

/**
 * excel中加载0f37数据
 *
 * @author wangshuai
 */
public class Load0F37DataServiceExcelImpl implements Load0F37DataService, Serializable {
    static final Logger logger = LoggerFactory.getLogger(Load0F37DataServiceExcelImpl.class);

    private String path = "";


    public String getPath() {
        return path;
    }

    public void setPath(String path) {
        this.path = path;
    }

    @Override
    public JavaRDD<Point> load(SparkSession spark, JavaSparkContext jsc, LoadDataParam param) {
        String dayStr = DateUtils.format(param.getDay(), DateUtils.DateFormat.YYYYMMDD);
        String type = "0f37";
        File file = new File(getPath() + "/" + type + "/" + dayStr + "/");
        try {
            List<Point> list = new ArrayList<>(10000);
            if (file.exists()) {
                File[] files = file.listFiles();
                for (File f : files) {
                    String name = f.getName();
                    if (StringUtils.endsWith(name, "xlsx")) {
                        List<Point> sublist = (List) ExcelUtils.readExcel(f, Point0F37.class);
                        list.addAll(sublist);
                    }
                }
            }
            return jsc.parallelize(list);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return null;

    }
}
